package com.lpssfxy.online.app

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.{DefaultJedisClientConfig, Jedis}

import scala.collection.JavaConverters._

// 定义一个连接助手对象，建立到redis和mongodb的连接
object ConnHelper extends Serializable {
  // 懒变量定义，使用的时候才初始化
  // 在每个节点上创建Jedis和MongoClient实例
  lazy val clientConfig = DefaultJedisClientConfig.builder()
    .password("openGauss_1234")
    .build()
  lazy val jedis = new Jedis("192.168.85.144", 6379, clientConfig)
  lazy val mongoClient = MongoClient(MongoClientURI("mongodb://fooadmin:123456@s3:27017/bigdata"))
}

// 定义连接mongo样例类
case class MongoConfig(uri: String, db: String)

// 定义标准推荐对象
case class Recommendation(productId: Int, score: Double)

// 定义商品相似度列表
case class ProductRecs(productId: Int, recs: Seq[Recommendation])

object RealtimeRecommender2018 {
  // 定义常量和表名
  val MONGODB_RATING_COLLECTION = "rating"
  val STREAM_RECS = "streamRecs"
  val PRODUCT_RECS = "productRecs"

  val MAX_USER_RATING_NUM = 20
  val MAX_SIM_PRODUCTS_NUM = 20

  // 常量提取（避免魔法值）
  val REDIS_HOST = "192.168.85.144"
  val REDIS_PORT = 6379
  val REDIS_PASSWORD = "openGauss_1234"
  val MONGO_URI = "mongodb://fooadmin:123456@s3:27017/bigdata"
  val MONGO_DB = "bigdata"
  val KAFKA_TOPIC = "recommender"
  val SPARK_CORES = "local[*]"

  def main(args: Array[String]): Unit = {

    // 创建spark conf
    val sparkConf = new SparkConf().setMaster(SPARK_CORES).setAppName(RealtimeRecommender2018.getClass.getSimpleName)
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(2))

    import spark.implicits._
    implicit val mongoConfig = MongoConfig(MONGO_URI, MONGO_DB)

    // 加载数据，相似度矩阵，广播出去
    val simProductsMatrix = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", PRODUCT_RECS)
      .format("com.mongodb.spark.sql")
      .load()
      .as[ProductRecs]
      .rdd
      // 为了后续查询相似度方便，把数据转换成map形式
      .map { item =>
        (item.productId, item.recs.map(x => (x.productId, x.score)).toMap)
      }
      .collectAsMap()
    // 定义广播变量
    val simProcutsMatrixBC = sc.broadcast(simProductsMatrix)

    // 创建kafka配置参数（核心修复点）
    val kafkaParam = Map(
      "bootstrap.servers" -> "s1:9092,s2:9092,s3:9092",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id" -> "recommender",
      "auto.offset.reset" -> "latest",
      "auto.commit.enable" -> "false"
    )

    // 创建 Kafka 流（参数类型严格匹配）
    val kafkaStream = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](
        Array(KAFKA_TOPIC),
        kafkaParam
      )
    )

    // 对kafkaStream进行处理，产生评分流，其数据格式为：userId|productId|score|timestamp
    val ratingStream = kafkaStream.flatMap { msg =>
      try {
        val attr = msg.value().split("\\|")
        Some((attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt))
      } catch {
        case e: Exception =>
          println(s"Invalid message format: ${msg.value()}, error: ${e.getMessage}")
          None // 过滤无效消息
      }
    }

    // 核心算法部分，定义评分流的处理流程
    ratingStream.foreachRDD { rdd =>
      rdd.foreachPartition { partition =>
        // 分区内创建连接（序列化安全）
        val redisClientConfig = DefaultJedisClientConfig.builder()
          .password(REDIS_PASSWORD)
          .build()
        val jedis = new Jedis(REDIS_HOST, REDIS_PORT, redisClientConfig)
        val mongoClient = MongoClient(MongoClientURI(MONGO_URI))
        implicit val mongoConfig = MongoConfig(MONGO_URI, MONGO_DB)

        try {
          partition.foreach {
            case (userId, productId, score, timestamp) =>
              // 业务逻辑（与原代码一致）
              println(">>>>>>>>>rating data coming!>>>>>>>>>>>>>>>>>>")

              // 1. 从redis里取出当前用户的最近评分，保存成一个数组Array[(productId, score)]
              val userRecentlyRatings = getUserRecentlyRatings(MAX_USER_RATING_NUM, userId, ConnHelper.jedis)
              println(">>>>>>>>>reading rating data from redis !>>>>>>>>>>>>>>>>>>")
              userRecentlyRatings.foreach(println)

              // 2. 从相似度矩阵中获取当前商品最相似的商品列表，作为备选列表，保存成一个数组Array[productId]
              val candidateProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM, productId, userId, simProcutsMatrixBC.value)
              println(">>>>>>>>>reading simProducts from productRecs in mongo!>>>>>>>>>>>>>>>>>>")
              candidateProducts.foreach(println)

              // 3. 计算每个备选商品的推荐优先级，得到当前用户的实时推荐列表，保存成 Array[(productId, score)]
              val streamRecs = computeProductScore(candidateProducts, userRecentlyRatings, simProcutsMatrixBC.value)
              println(">>>>>>>>>computing product score and getting streamRecs!>>>>>>>>>>>>>>>>>>")
              streamRecs.foreach(println)

              // 4. 存入mongodb
              saveDataToMongoDB(userId, streamRecs, mongoClient)

          }
        } finally {
          // 分区处理完成后必须关闭连接（重要！避免资源泄漏）
          jedis.close()
          mongoClient.close()
        }
      }
    }

    // 启动streaming
    ssc.start()
    println("streaming started!")
    ssc.awaitTermination()
  }

  /**
   * 从redis里获取最近num次评分
   */

  import scala.collection.JavaConversions._

  def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis): Array[(Int, Double)] = {
    // 从redis中用户的评分队列里获取评分数据，list键名为uid:USERID，值格式是 PRODUCTID:SCORE
    jedis.lrange("userId:" + userId.toString, 0, num)
      .map { item =>
        val attr = item.split("\\:")
        (attr(0).trim.toInt, attr(1).trim.toDouble)
      }
      .toArray
  }

  // 获取当前商品的相似列表，并过滤掉用户已经评分过的，作为备选列表
  def getTopSimProducts(num: Int,
                        productId: Int,
                        userId: Int,
                        simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                       (implicit mongoConfig: MongoConfig): Array[Int] = {
    // 从广播变量相似度矩阵中拿到当前商品的相似度列表
    val allSimProducts = simProducts(productId).toArray
    // 获得用户已经评分过的商品，过滤掉，排序输出
    val ratingCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
    val ratingExist = ratingCollection.find(MongoDBObject("userId" -> userId))
      .toArray
      .map { item => // 只需要productId
        item.get("productId").toString.toInt
      }
    // 从所有的相似商品中进行过滤
    allSimProducts.filter(x => !ratingExist.contains(x._1))
      .sortWith(_._2 > _._2)
      .take(num)
      .map(x => x._1)
  }

  // 计算每个备选商品的推荐得分
  def computeProductScore(candidateProducts: Array[Int],
                          userRecentlyRatings: Array[(Int, Double)],
                          simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
  : Array[(Int, Double)] = {
    // 定义一个长度可变数组ArrayBuffer，用于保存每一个备选商品的基础得分，(productId, score)
    val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
    // 定义两个map，用于保存每个商品的高分和低分的计数器，productId -> count
    val increMap = scala.collection.mutable.HashMap[Int, Int]()
    val decreMap = scala.collection.mutable.HashMap[Int, Int]()

    // 遍历每个备选商品，计算和已评分商品的相似度
    for (candidateProduct <- candidateProducts; userRecentlyRating <- userRecentlyRatings) {
      // 从相似度矩阵中获取当前备选商品和当前已评分商品间的相似度
      val simScore = getProductsSimScore(candidateProduct, userRecentlyRating._1, simProducts)
      if (simScore > 0.4) {
        // 按照公式进行加权计算，得到基础评分
        scores += ((candidateProduct, simScore * userRecentlyRating._2))
        if (userRecentlyRating._2 > 3) {
          increMap(candidateProduct) = increMap.getOrDefault(candidateProduct, 0) + 1
        } else {
          decreMap(candidateProduct) = decreMap.getOrDefault(candidateProduct, 0) + 1
        }
      }
    }

    // 根据公式计算所有的推荐优先级，首先以productId做groupby
    scores.groupBy(_._1).map {
        case (productId, scoreList) =>
          (productId, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(productId, 1)) - log(decreMap.getOrDefault(productId, 1)))
      }
      // 返回推荐列表，按照得分排序
      .toArray
      .sortWith(_._2 > _._2)
  }

  def getProductsSimScore(product1: Int, product2: Int,
                          simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Double = {
    simProducts.get(product1) match { // 模式匹配
      case Some(sims) => sims.get(product2) match { // 二次模式匹配
        case Some(score) => score // 匹配到了直接返回score
        case None => 0.0 // 没有匹配到直接返回0.0
      }
      case None => 0.0
    }
  }

  // 自定义log函数，以N为底，实现了换底公式
  def log(m: Int): Double = {
    val N = 10
    math.log(m) / math.log(N)
  }

  // 修改 saveDataToMongoDB 方法，接收 mongoClient
  def saveDataToMongoDB(userId: Int, streamRecs: Array[(Int, Double)], mongoClient: MongoClient)(implicit mongoConfig: MongoConfig): Unit = {
    val streamRecsCollection = mongoClient(mongoConfig.db)(STREAM_RECS)
    streamRecsCollection.findAndRemove(MongoDBObject("userId" -> userId))
    streamRecsCollection.insert(MongoDBObject("userId" -> userId, "recs" -> streamRecs.map(x => MongoDBObject("productId" -> x._1, "score" -> x._2))))
  }
}